package org.gjy.m8.thread;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author 宫静雨
 * @version 1.0
 * @since 2023-09-06 15:31:08
 */
public class BlockingQueueTest {

    private static final Logger log = LoggerFactory.getLogger(BlockingQueueTest.class);

    @Test
    public void test1() throws InterruptedException {
        int threads = 20;
        DataBuffer<String> data = new DataBuffer<>(100);
        CountDownLatch p = new CountDownLatch(threads);
        CountDownLatch c = new CountDownLatch(threads);
        LongAdder producerAdder = new LongAdder();
        LongAdder consumerAdder = new LongAdder();

        Callable<String> producer = () -> {
            String s = UUID.randomUUID().toString();
            data.add(s);
            log.info("producer: {}, {}", s, LocalDateTime.now());
            p.countDown();
            producerAdder.increment();
            return s;
        };

        Callable<String> consumer = () -> {
            String s = data.fetch();
            log.info("consumer: {}, {}", s, LocalDateTime.now());
            c.countDown();
            consumerAdder.increment();
            return s;
        };

        ExecutorService pool = Executors.newFixedThreadPool(threads);
        for (int i = 0; i < threads; i++) {
            pool.submit(producer);
        }
        for (int i = 0; i < threads; i++) {
            pool.submit(consumer);
        }

        p.await();
        c.await();

        log.info("producerAdder {}", producerAdder);
        log.info("consumerAdder {}", consumerAdder);
    }

    private static class DataBuffer<T> {
        private final BlockingQueue<T> queue;

        public DataBuffer(Integer max) {
            queue = new ArrayBlockingQueue<>(max);
        }

        public void add(T e) {
            queue.add(e);
        }

        public T fetch() throws Exception {
            return queue.take();
        }
    }

}
